
package com.chinacreator.c2etcd;


import com.google.protobuf.ByteString;
import io.etcd.jetcd.*;
import io.etcd.jetcd.cluster.MemberListResponse;
import io.etcd.jetcd.kv.GetResponse;
import io.etcd.jetcd.kv.PutResponse;
import io.etcd.jetcd.lease.LeaseGrantResponse;
import io.etcd.jetcd.lease.LeaseKeepAliveResponse;
import io.etcd.jetcd.lease.LeaseRevokeResponse;
import io.etcd.jetcd.lease.LeaseTimeToLiveResponse;
import io.etcd.jetcd.lock.LockResponse;
import io.etcd.jetcd.lock.UnlockResponse;
import io.etcd.jetcd.maintenance.AlarmMember;
import io.etcd.jetcd.maintenance.AlarmResponse;
import io.etcd.jetcd.maintenance.StatusResponse;
import io.etcd.jetcd.options.*;
import io.etcd.jetcd.watch.WatchEvent;
import io.etcd.jetcd.watch.WatchResponse;
import io.grpc.stub.StreamObserver;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.URI;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;

import static com.chinacreator.c2etcd.EtcdChangedEvent.Type.*;
import static com.google.protobuf.ByteString.copyFromUtf8;
import static io.etcd.jetcd.ByteSequence.from;

/**
 * @Author: jack.liang
 **/
public class EtcdClientImpl implements EtcdClient {

    private static final Logger LOGGER = LoggerFactory.getLogger(EtcdClientImpl.class);

    private Client client;

    private KV kvClient;

    private Lease leaseClient;

    private Lock lockClient;
    // TODO: 2020/2/7 新版JETCD已经更新，需要从新基于stub grpc开发watch
    private Watch watchClient;

    private Cluster clusterClient;

    private Maintenance maintenanceClient;

    private HeartBeatLease heartBeatLease;

    private HeartBeatLease heartBeatLeaseElectionLeader;

    private volatile ConcurrentHashMap<String, EtcdWatcher> etcdWatchers = new ConcurrentHashMap<>();

    //	Use synchronous waiting mechanism to avoid delay caused by network problems ->  As countDownLatcher
    private static final String ETCD_BASE_COUNTDOWN_LATCHER = "__base__v5";

    /**
     * $EtcdClientImpl
     * <B>构造方法</B>Etcd Client 无心跳租约<BR>
     *
     * @param address 服务地址
     */
    public EtcdClientImpl(String address) {
        this(address, false, false,  null, null, null);
    }

    /**
     * $EtcdClientImpl
     * <B>构造方法</B>Etcd Client 可配置心跳租约<BR>
     *
     * @param address            服务地址
     * @param usedHeartBeatLease 是否启用心跳租约机制
     */
    public EtcdClientImpl(String address, boolean usedHeartBeatLease) {
        this(address, usedHeartBeatLease, false,  null, null, null);
    }

    /**
     * $EtcdClientImpl
     * <B>构造方法</B>Etcd Client 可配置心跳租约, 懒加载, 负载均衡策略<BR>
     *
     * @param address            服务地址
     * @param usedHeartBeatLease 是否启用心跳租约机制
     * @param lazyInitialization 是否启用懒加载
     */
    public EtcdClientImpl(String address, boolean usedHeartBeatLease, boolean lazyInitialization) {
        this(address, usedHeartBeatLease, lazyInitialization, null, null, null);
    }

    /**
     * $EtcdClientImpl
     * <B>构造方法</B>Etcd Client 自定义配置<BR>
     *
     * @param address            服务地址
     * @param usedHeartBeatLease 是否启用心跳租约机制
     * @param lazyInitialization 是否启用懒加载机制
     * @param authority          添加认证信息
     * @param username           用户名
     * @param password           密码
     */
    public EtcdClientImpl(String address, boolean usedHeartBeatLease, boolean lazyInitialization, String authority, String username, String password) {
        ClientBuilder clientBuilder = Client.builder();
        String[] addressList = address.split(",");
        clientBuilder.lazyInitialization(lazyInitialization).endpoints(addressList);
//        if (!loadBalancer) {
//            clientBuilder.loadBalancerFactory(PickFirstBalancerFactory.getInstance());
//        } else {
//            clientBuilder.loadBalancerFactory(RoundRobinLoadBalancerFactory.getInstance());
//        }
        if (StringUtils.isNoneBlank(authority) && StringUtils.isNoneBlank(username) && StringUtils.isNoneBlank(password)) {
            clientBuilder.authority(authority);
            clientBuilder.user(from(username.getBytes()));
            clientBuilder.password(from(password.getBytes()));
        }
        this.client = clientBuilder.build();
        this.kvClient = client.getKVClient();
        this.leaseClient = client.getLeaseClient();
        this.watchClient = client.getWatchClient();
        this.lockClient = client.getLockClient();
        this.clusterClient = client.getClusterClient();
        this.maintenanceClient = client.getMaintenanceClient();

        try {
            // first as countDownLatch init
            KeyValue kv = getKey(ETCD_BASE_COUNTDOWN_LATCHER);
            if (kv == null) {
                putKey(ETCD_BASE_COUNTDOWN_LATCHER, ETCD_BASE_COUNTDOWN_LATCHER);
            }
            LOGGER.info("#EtcdClientImpl#Constructor Init ok!");
            //	second used heatbeat lease
            if (usedHeartBeatLease) {
                this.heartBeatLease = new HeartBeatLease(EtcdClientImpl.this);
                this.heartBeatLease.start();
                LOGGER.info("#EtcdClientImpl.HeartBeatLease# heartbeat lease thread is running!");
            }
        } catch (Exception e) {
            LOGGER.error("#EtcdClientImpl#Constructor Init Error! ", e);
        } finally {
            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                try {
                    this.close();
                    LOGGER.info("#EtcdClientImpl# EtcdClient Shutdown Hook ok!");
                } catch (InterruptedException | ExecutionException e) {
                    LOGGER.error("#EtcdClientImpl# EtcdClient Shutdown Hook error!", e);
                }
            }));
        }
    }

    /**
     * <B>方法名称：</B>putKey<BR>
     * <B>概要说明：</B><BR>
     *
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public void putKey(String key, String value) throws Exception {
        kvClient.put(from(copyFromUtf8(key)), from(value.getBytes(CHARSET)));
    }

    /**
     * <B>方法名称：</B>getKey<BR>
     * <B>概要说明：</B><BR>
     *
     * @return
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public KeyValue getKey(final String key) throws Exception {
        GetResponse getResponse = kvClient.get(from(key, Charset.isSupported("UTF-8") ? Charset.forName("UTF-8") : Charset.defaultCharset())).get();
        if (getResponse == null) {
            return null;
        }
        List<KeyValue> kvs = getResponse.getKvs();
        if (kvs.size() > 0) {
            return kvs.get(0);
        } else {
            return null;
        }
    }

    /**
     * <B>方法名称：</B>deleteKey<BR>
     * <B>概要说明：</B><BR>
     *
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public void deleteKey(String key) {
        kvClient.delete(from(copyFromUtf8(key)));
    }

    /**
     * <B>方法名称：</B>getKeyWithPrefix<BR>
     * <B>概要说明：</B><BR>
     *
     * @return
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public List<KeyValue> getKeyWithPrefix(String prefix) {
        List<KeyValue> keyValues = new ArrayList<>();
        GetOption getOption = GetOption.
                newBuilder().
                withPrefix(from(copyFromUtf8(prefix))).
                build();
        try {
            keyValues = kvClient.
                    get(from(copyFromUtf8(prefix)), getOption).
                    get().
                    getKvs();

        } catch (Exception e) {
            e.printStackTrace();
        }
        return keyValues;
    }

    /**
     * <B>方法名称：</B>deleteKeyWithPrefix<BR>
     * <B>概要说明：</B><BR>
     *
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public void deleteKeyWithPrefix(String prefix) {
        DeleteOption deleteOption = DeleteOption.newBuilder().withPrefix(from(copyFromUtf8(prefix))).build();
        kvClient.delete(from(copyFromUtf8(prefix)), deleteOption);
    }

    /**
     * <B>方法名称：</B>putKeyWithPrivateLease<BR>
     * <B>概要说明：</B>添加KeyValue, 并为其设置指定的私有租约putKeyWithPrivateLease<BR>
     *
     * @param key        唯一键
     * @param onNext     消费行为
     * @return long
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public void addWatcherListener(String key, Consumer<WatchResponse> onNext) {
            watchClient.watch(from(copyFromUtf8(key)),onNext);
    }

    /**
     * <B>方法名称：</B>putKeyWithPrivateLease<BR>
     * <B>概要说明：</B>添加KeyValue, 并为其设置指定的私有租约putKeyWithPrivateLease<BR>
     *
     * @param key        唯一键
     * @param value      内容
     * @param expireTime 过期时间
     * @return long
     * @author jack.liang
     * @date 2020/2/7
     */
    private long putKeyWithPrivateLease(String key, String value, long expireTime) {
        CompletableFuture<LeaseGrantResponse> leaseGrantResponse = leaseClient.grant(expireTime);
        PutOption putOption;
        try {
            putOption = PutOption.newBuilder().withLeaseId(leaseGrantResponse.get().getID()).build();
            kvClient.put(from(ByteString.copyFromUtf8(key)), from(ByteString.copyFromUtf8(value)), putOption);
            return leaseGrantResponse.get().getID();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return 0L;
    }

    /**
     * <B>方法名称：</B>putKeyWithPublicLease<BR>
     * <B>概要说明：</B>添加KeyValue, 并为其设置指定的共有租约putKeyWithPublicLease<BR>
     *
     * @param key     唯一键
     * @param value   内容
     * @param leaseId 公有租约ID
     * @return long
     * @throws Exception
     * @author jack.liang
     * @date 2020/2/7
     */
    private long putKeyWithPublicLease(String key, String value, long leaseId) {
        PutOption putOption = PutOption.newBuilder().withLeaseId(leaseId).build();
        CompletableFuture<PutResponse> putResponse = kvClient.put(from(ByteString.copyFromUtf8(key)), from(copyFromUtf8(value)), putOption);
        try {
            return putResponse.get().getHeader().getRevision();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return 0L;
    }

    /**
     * <B>方法名称：</B>putKeyWithExpireTime<BR>
     * <B>概要说明：</B>添加KeyValue, 并为其设置过期时间<BR>
     *
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public long putKeyWithExpireTime(String key, String value, long expireTime) {
        return putKeyWithPrivateLease(key, value, expireTime);
    }

    /**
     * <B>方法名称：</B>putKeyWithLeaseId<BR>
     * <B>概要说明：</B>添加KeyValue, 并为其设置指定的租约ID, 统一租期时间<BR>
     *
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public long putKeyWithLeaseId(String key, String value, long leaseId) throws Exception {
        return putKeyWithPublicLease(key, value, leaseId);
    }

    /**
     * <B>方法名称：</B>generatorLeaseId<BR>
     * <B>概要说明：</B>生成租约<BR>
     *
     * @return 返回long类型租约ID
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public long generatorLeaseId(long expireTime) throws Exception {
        CompletableFuture<LeaseGrantResponse> leaseGrantResponse = leaseClient.grant(expireTime);
        return leaseGrantResponse.get().getID();
    }

    /**
     * <B>方法名称：</B>keepAliveSingleLease<BR>
     * <B>概要说明：</B>保持租约<BR>
     *
     * @return KeepAliveListener
     * todo
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public CloseableClient keepAliveSingleLease(long leaseId) {
        CloseableClient closeableClient = leaseClient.keepAlive(leaseId, new StreamObserver<LeaseKeepAliveResponse>() {

            @Override
            public void onNext(LeaseKeepAliveResponse leaseKeepAliveResponse) {
                LOGGER.info("Ttl leaseKeepAliveResponse  == {} lease == {} ",
                        leaseKeepAliveResponse.getTTL(),
                        leaseKeepAliveResponse.toString());
            }

            @Override
            public void onError(Throwable throwable) {
                //ignore
            }

            @Override
            public void onCompleted() {
                //ignore
            }
        });
        return closeableClient;
    }

    /**
     * <B>方法名称：</B>keepAliveOnce<BR>
     * <B>概要说明：</B>单次租约<BR>
     *
     * @param leaseId 租约ID
     * @return LeaseKeepAliveResponse
     * @throws ExecutionException
     * @throws InterruptedException
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public LeaseKeepAliveResponse keepAliveOnce(long leaseId) throws InterruptedException, ExecutionException {
        CompletableFuture<LeaseKeepAliveResponse> completableFuture = leaseClient.keepAliveOnce(leaseId);
        if (completableFuture != null) {
            return completableFuture.get();
        }
        return null;
    }

    /**
     * <B>方法名称：</B>timeToLiveLease<BR>
     * <B>概要说明：</B>请求租约timeToLive API返回给定租约ID的附加密钥<BR>
     *
     * @param leaseId 租约ID
     * @return LeaseTimeToLiveResponse
     * @throws ExecutionException
     * @throws InterruptedException
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public LeaseTimeToLiveResponse timeToLiveLease(long leaseId) throws InterruptedException, ExecutionException {
        CompletableFuture<LeaseTimeToLiveResponse> completableFuture = leaseClient.timeToLive(leaseId, LeaseOption.newBuilder().withAttachedKeys().build());
        if (completableFuture != null) {
            return completableFuture.get();
        }
        return null;
    }

    /**
     * <B>方法名称：</B>revokeLease<BR>
     * <B>概要说明：</B>回收租约ID<BR>
     *
     * @param leaseId 租约ID
     * @return LeaseRevokeResponse
     * @throws ExecutionException
     * @throws InterruptedException
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public LeaseRevokeResponse revokeLease(long leaseId) throws InterruptedException, ExecutionException {
        CompletableFuture<LeaseRevokeResponse> completableFuture = leaseClient.revoke(leaseId);
        if (completableFuture != null) {
            return completableFuture.get();
        }
        return null;
    }

    /**
     * <B>方法名称：</B>getHeartBeatLeaseId<BR>
     * <B>概要说明：</B>获取心跳全局租约ID<BR>
     *
     * @return if used HeartBeatLease return LeaseId , else return 0;
     * @throws InterruptedException
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public long getHeartBeatLeaseId() throws InterruptedException {
        return heartBeatLease != null ? heartBeatLease.getLeaseId() : 0L;
    }

    /**
     * <B>方法名称：</B>lock<BR>
     * <B>概要说明：</B>使用分布式锁<BR>
     *
     * @param lockName 锁名称
     * @return LockResponse
     * @throws Exception
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public LockResponse lock(String lockName) throws Exception {
        CompletableFuture<LockResponse> completableFuture = lockClient.lock(from(copyFromUtf8(lockName)), 0);
        if (completableFuture != null) {
            return completableFuture.get();
        }
        return null;
    }

    /**
     * <B>方法名称：</B>lock<BR>
     * <B>概要说明：</B>使用分布式锁，带租期时间单位是秒<BR>
     *
     * @param lockName   锁名称
     * @param expireTime 带租期时间单位是秒
     * @return LockResponse
     * @throws Exception
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public LockResponse lock(String lockName, long expireTime) throws Exception {
        CompletableFuture<LeaseGrantResponse> leaseGrantResponse = leaseClient.grant(expireTime);
        CompletableFuture<LockResponse> completableFuture = lockClient.lock(from(copyFromUtf8(lockName)), leaseGrantResponse.get().getID());
        if (completableFuture != null) {
            return completableFuture.get();
        }
        return null;
    }

    /**
     * <B>方法名称：</B>unlock<BR>
     * <B>概要说明：</B>根据名称释放锁<BR>
     *
     * @param lockName 名称
     * @return UnlockResponse
     * @throws Exception
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public UnlockResponse unlock(String lockName) throws Exception {
        CompletableFuture<UnlockResponse> completableFuture = lockClient.unlock(from(copyFromUtf8(lockName)));
        if (completableFuture != null) {
            return completableFuture.get();
        }
        return null;
    }

//    /**
//     * <B>方法名称：</B>watch<BR>
//     * <B>概要说明：</B>指定watch的key,使用拓展<BR>
//     *
//     * @param key      key 名字
//     * @param listener 监听器
//     * @return
//     */
//    @Override
//    public Watch.Watcher watchKeyCallback(String key, WatchOption option, Watch.Listener listener) {
//        return watchClient.watch(ByteSequence.from(copyFromUtf8(key)), option == null ? WatchOption.DEFAULT : option, listener);
//    }

    /**
     * <B>方法名称：</B>lockByLeaseId<BR>
     * <B>概要说明：</B>使用分布式锁，带租期id<BR>
     *
     * @param lockName 锁名称
     * @param leaseId  租期id
     * @return LockResponse
     * @throws Exception
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public LockResponse lockByLeaseId(String lockName, long leaseId) throws Exception {
        CompletableFuture<LockResponse> completableFuture = lockClient.lock(from(copyFromUtf8(lockName)), leaseId);
        if (completableFuture != null) {
            return completableFuture.get();
        }
        return null;
    }

    /**
     * <B>方法名称：</B>lockByElectionLeader<BR>
     * <B>概要说明：</B>用于分布式选举业务Leader, 全局方法, 单次生效, 所有followers调用该方法竞选业务Leader, 失败则阻塞等待<BR>
     * <B>注意事项：</B>注意该Leader为业务Leader, 并不是指代etcd集群Leader<BR>
     *
     * @param lockName 名称
     * @return LockResponse
     * @throws Exception
     * @author jack.liang
     * @date 2020/2/7
     */
    @Deprecated
    @Override
    public synchronized LockResponse lockByElectionLeader(String lockName) throws Exception {
        if (this.heartBeatLeaseElectionLeader == null) {
            this.heartBeatLeaseElectionLeader = new HeartBeatLease(EtcdClientImpl.this);
            this.heartBeatLeaseElectionLeader.start();
            LOGGER.info("#EtcdClientImpl.lockByElectionLeader# heartbeat thread is running, lockName : {}", lockName);
        }
        long leaseId = this.heartBeatLeaseElectionLeader.getLeaseId();
        LockResponse lockResponse = lockByLeaseId(lockName, leaseId);
        return lockResponse;
    }

    /**
     * $HeartBeatLease
     *
     * @author jack.liang
     * @date 2020/2/7
     */
    private class HeartBeatLease implements Runnable {

        private EtcdClient etcdClient;

        private ScheduledExecutorService heartBeat;

        private final static long DEFAULT_LEASE_TIME = 13;

        private final static long DEFAULT_INITIALDELAY = 10;

        private final static long DEFAULT_PERIOD = 4000;

        private volatile long leaseTime;

        private volatile long initialDelay;

        private volatile long period;

        private long leaseId;

        private AtomicBoolean isFirstHeartBeat = new AtomicBoolean(true);

        public HeartBeatLease(EtcdClient etcdClient) {
            this(etcdClient, DEFAULT_LEASE_TIME, DEFAULT_INITIALDELAY, DEFAULT_PERIOD);
        }

        public HeartBeatLease(EtcdClient etcdClient, long leaseTime, long initialDelay, long period) {
            this.etcdClient = etcdClient;
            this.leaseTime = leaseTime;
            this.initialDelay = initialDelay;
            this.period = period;
        }

        public void start() throws InterruptedException {
            this.heartBeat = Executors.newScheduledThreadPool(1);
            this.heartBeat.scheduleAtFixedRate(HeartBeatLease.this, initialDelay, period, TimeUnit.MILLISECONDS);
        }

        @Override
        public void run() {
            try {
                if (isFirstHeartBeat.get()) {
                    //暂时先不动,这里是有个死循环问题
                    leaseId = etcdClient.generatorLeaseId(leaseTime);
                    isFirstHeartBeat.compareAndSet(true, false);
                }
                etcdClient.keepAliveOnce(leaseId);
            } catch (Throwable t) {
                LOGGER.error("#EtcdClientImpl.HeartBeatLease# heartbeat lease Error, leaseId : {}", leaseId, t);
            }
        }

        /**
         * <B>方法名称：</B>stop<BR>
         * <B>概要说明：</B>revokeLease and shutdown threadpool<BR>
         * // 暂时先不动
         *
         * @throws InterruptedException
         * @throws ExecutionException
         * @author jack.liang
         * @date 2020/2/7
         */
        public void stop() throws InterruptedException, ExecutionException {
            this.etcdClient.revokeLease(leaseId);
            this.heartBeat.shutdown();
        }

        /**
         * <B>方法名称：</B>getLeaseId<BR>
         * <B>概要说明：</B>获取心跳租约ID<BR>
         * * 暂时先不动
         *
         * @return LeaseId
         * @throws InterruptedException
         * @author jack.liang
         * @date 2020/2/7
         */
        public long getLeaseId() throws InterruptedException {
            for (; ; ) {
                if (!isFirstHeartBeat.get()) {
                    break;
                }
            }
            return leaseId;
        }

    }

//    /**
//     * <B>方法名称：</B>addWatcherListener<BR>
//     * <B>概要说明：</B>添加watcher监听<BR>
//     *
//     * @param watcherKey      根据监听Key
//     * @param usePrefix       是否启用前缀监听
//     * @param watcherListener 监听回调对象
//     * @author jack.liang
//     * @date 2020/2/7
//     */
//    @Override
//    public synchronized void addWatcherListener(final String watcherKey, final boolean usePrefix, WatcherListener watcherListener) {
//        EtcdWatcher etcdWatcher = etcdWatchers.get(watcherKey);
//        if (etcdWatcher == null) {
//            etcdWatcher = new EtcdWatcher(EtcdClientImpl.this, watcherKey, usePrefix, watcherListener);
//            etcdWatcher.start();
//            etcdWatchers.putIfAbsent(watcherKey, etcdWatcher);
//            LOGGER.info("addWatcherListener watcherKey : {}, usePrefix : {}", watcherKey, usePrefix);
//        }
//    }
//
//    /**
//     * <B>方法名称：</B>removeWatcherListener<BR>
//     * <B>概要说明：</B>移除watcher监听<BR>
//     *
//     * @param watcherKey 根据监听Key
//     * @author jack.liang
//     * @date 2020/2/7
//     */
//    @Override
//    public synchronized void removeWatcherListener(final String watcherKey) {
//        EtcdWatcher etcdWatcher = etcdWatchers.get(watcherKey);
//        if (etcdWatcher != null) {
//            etcdWatcher.stop();
//            etcdWatchers.remove(etcdWatcher.getWatcherKey());
//            LOGGER.info("removeWatcherListener watcherKey : {}", etcdWatcher.getWatcherKey());
//        }
//    }

    /**
     * <B>方法名称：</B>close<BR>
     * <B>概要说明：</B>release all resources<BR>
     *
     * @throws ExecutionException
     * @throws InterruptedException
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public void close() throws InterruptedException, ExecutionException {
        for(Map.Entry<String, EtcdWatcher> me : etcdWatchers.entrySet()){
            EtcdWatcher etcdWatcher = me.getValue();
            etcdWatcher.stop();
        }
        if (heartBeatLease != null) {
            heartBeatLease.stop();
        }
        if (heartBeatLeaseElectionLeader != null) {
            heartBeatLeaseElectionLeader.stop();
        }
        if (client != null) {
            client.close();
        }
    }


    /**
     * <B>方法名称：</B>listMember<BR>
     * <B>概要说明：</B>获取集群成员<BR>
     *
     * @return MemberListResponse
     * @throws ExecutionException
     * @throws InterruptedException
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public MemberListResponse listMember() throws InterruptedException, ExecutionException {
        CompletableFuture<MemberListResponse> completableFuture = this.clusterClient.listMember();
        if (completableFuture != null) {
            return completableFuture.get();
        }
        return null;
    }


    /**
     * $EtcdWatcher
     *
     * @author jack.liang
     * @date 2020/2/7
     */
    @Deprecated
    private class EtcdWatcher {

        private volatile boolean isRunning = true;

        private ExecutorService watcherExecutor;

        private EtcdClient etcdClient;

        private String watcherKey;
        //v3.0 escape for
        private Watch.Watcher watcher;

        private AtomicReference<WatchResponse> ref = new AtomicReference<>();

        private WatcherListener watcherListener;

        public EtcdWatcher(EtcdClient etcdClient, final String watcherKey, final boolean usePrefix, WatcherListener watcherListener) {
            this.etcdClient = etcdClient;
            this.watcherExecutor = Executors.newSingleThreadExecutor(new NamedThreadFactory("etcd-watcher-", true));
            this.watcherKey = watcherKey;
            if (usePrefix) {
                watcher = watchClient.watch(from(copyFromUtf8(watcherKey)),
                        WatchOption.newBuilder().withPrefix(from(ByteString.copyFromUtf8(watcherKey))).build(),
                        ref::set);
            } else {
                watcher = watchClient.watch(from(copyFromUtf8(watcherKey)), ref::set);
            }
            this.watcherListener = watcherListener;
        }

        public void start() {
            this.watcherExecutor.submit(new Callable<Object>() {

                @Override
                public Object call() throws Exception {

                    while (isRunning) {
                        try {
                            List<WatchEvent> watcherList =  ref.get().getEvents();
                            for (WatchEvent watchEvent : watcherList) {
                                KeyValue keyValue = watchEvent.getKeyValue();
                                switch (watchEvent.getEventType()) {
                                    case PUT:
                                        watcherListener.watcherKeyChanged(etcdClient, new EtcdChangedEvent(
                                                keyValue.getKey().toString(),
                                                keyValue.getValue().toString(),
                                                PUT));
                                        break;
                                    case DELETE:
                                        watcherListener.watcherKeyChanged(etcdClient, new EtcdChangedEvent(
                                                keyValue.toString(),
                                                keyValue.toString(),
                                                DELETE));
                                        break;
                                    case UNRECOGNIZED:
                                        // ignore
                                        LOGGER.warn("#EtcdClientImpl.EtcdWatcher# watched UNRECOGNIZED Warn, Type: {} ", UNRECOGNIZED);
//										watcherListener.watcherKeyChanged(etcdClient, new EtcdChangedEvent(
//												keyValue.getKey().toStringUtf8(),
//												keyValue.getValue().toStringUtf8(),
//												EtcdChangedEvent.Type.UNRECOGNIZED));
                                        break;
                                    default:
                                        break;
                                }
                            }
                        } catch (InterruptedException e) {
                            LOGGER.warn("#EtcdClientImpl.EtcdWatcher# watcher running thread is Warn, catch InterruptedException! ", e);
                            // ignore
                        } catch (Throwable e) {
                            LOGGER.error("#EtcdClientImpl.EtcdWatcher# watcher running thread is Error, catch Throwable! ", e);
                            // ignore
                        }
                    }
                    return null;
                }
            });
        }

        public void stop() {
            this.isRunning = false;
            this.watcherExecutor.shutdown();
        }

        public String getWatcherKey() {
            return watcherKey;
        }

    }


    /**
     * <B>方法名称：</B>statusMember<BR>
     * <B>概要说明：</B>查看某一个成员的状态<BR>
     *
     * @param endpoint 端点地址
     * @return StatusResponse
     * @throws ExecutionException
     * @throws InterruptedException
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public StatusResponse statusMember(String endpoint) throws InterruptedException, ExecutionException {
        CompletableFuture<StatusResponse> completableFuture = this.maintenanceClient.statusMember(URI.create(endpoint));
        if (completableFuture != null) {
            return completableFuture.get();
        }
        return null;
    }


    /**
     * <B>方法名称：</B>statusMember<BR>
     * <B>概要说明：</B>解除警报<BR>
     *
     * @param member 指定成员
     * @return AlarmResponse
     * @throws ExecutionException
     * @throws InterruptedException
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public AlarmResponse alarmDisarm(AlarmMember member) throws InterruptedException, ExecutionException {
        CompletableFuture<AlarmResponse> completableFuture = this.maintenanceClient.alarmDisarm(member);
        if (completableFuture != null) {
            return completableFuture.get();
        }
        return null;
    }

    /**
     * <B>方法名称：</B>listAlarms<BR>
     * <B>概要说明：</B>获取所有的报警信息<BR>
     *
     * @return AlarmResponse
     * @throws ExecutionException
     * @throws InterruptedException
     * @author jack.liang
     * @date 2020/2/7
     */
    @Override
    public AlarmResponse listAlarms() throws InterruptedException, ExecutionException {
        CompletableFuture<AlarmResponse> completableFuture = this.maintenanceClient.listAlarms();
        if (completableFuture != null) {
            return completableFuture.get();
        }
        return null;
    }
}

